package org.big.kafkacluster;

import org.apache.kafka.clients.admin.*;

import java.util.*;
import java.util.concurrent.ExecutionException;

//使用Java管理Topic
public class KafkaTopicCRUD {

    private static AdminClient getAdminClient() {
        Properties props = new Properties();
        props.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        return AdminClient.create(props);
    }

    public static void createTopic(String topicName, int numPartitions, short replicationFactor) throws ExecutionException, InterruptedException {
        try (AdminClient adminClient = getAdminClient()) {
            // 定义要创建的新Topic
            NewTopic newTopic = new NewTopic(topicName, numPartitions, replicationFactor);
            // 执行创建操作
            CreateTopicsResult result = adminClient.createTopics(Collections.singletonList(newTopic));
            // 等待创建完成（同步）
            result.all().get();
            System.out.println("Topic '" + topicName + "' created successfully.");
        }
    }

    public static void listTopics() throws ExecutionException, InterruptedException {
        try (AdminClient adminClient = getAdminClient()) {
            // 列出集群中的所有Topic
            ListTopicsResult result = adminClient.listTopics();
            Set<String> topicNames = result.names().get();
            System.out.println("Existing topics: " + topicNames);
        }
    }

    public static void describeTopic(String topicName) throws ExecutionException, InterruptedException {
        try (AdminClient adminClient = getAdminClient()) {
            // 查看指定Topic的详细信息（分区、副本等）
            DescribeTopicsResult result = adminClient
                    .describeTopics(Collections.singletonList(topicName));
            TopicDescription description = result.topicNameValues().get(topicName).get();
            System.out.println("Description of '" + topicName + "': " + description);
        }
    }

    public static void deleteTopic(String topicName) throws ExecutionException, InterruptedException {
        try (AdminClient adminClient = getAdminClient()) {
            // 删除指定的Topic（谨慎操作！）
            DeleteTopicsResult result = adminClient.deleteTopics(Collections.singletonList(topicName));
            result.all().get();
            System.out.println("Topic '" + topicName + "' deleted successfully.");
        }
    }

    public static void main(String[] args) throws Exception {
        String myTopic = "my-learning-topic";

        // 1. 创建Topic（3个分区，1个副本因子）
        createTopic(myTopic, 3, (short) 1);

        // 2. 列出所有Topic
        listTopics();

        // 3. 查看Topic详情
        describeTopic(myTopic);

        // 4. 删除Topic（取消注释以执行，数据将丢失！）
        // deleteTopic(myTopic);
    }
}